package com.whq.thrift.api.common.thriftserver;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.whq.common.StringUtil;
import com.whq.common.UUIDUtil;
import com.whq.thrift.api.common.constants.CommonConfigConstants;
import com.whq.thrift.api.common.constants.CommonErrorCodeConstants;
import com.whq.thrift.api.common.kafka.KafkaStringProducer;
import com.whq.thrift.api.common.thriftclient.AbstractApiJsonCallback;
import com.whq.thrift.api.common.thriftclient.ThriftApiJsonClientHelper;

public abstract class AbstractApiJsonService extends AbstractBaseService  {
	private static final Logger LOGGER = LoggerFactory.getLogger(AbstractApiJsonService.class);
	
	private static final Logger LOGGER_THRIFT_CONSUME = LoggerFactory.getLogger(CommonConfigConstants.LOGGER_KEY_THRIFT_CONSUME);
	
	private static final Logger LOGGER_THRIFT_REQUEST_RESPONSE = LoggerFactory.getLogger(CommonConfigConstants.LOGGER_KEY_THRIFT_REQUEST_RESPONSE);
	
	private String thriftConsumeLogKafkaTopic = "";
	private ThriftApiJsonClientHelper thriftApiJsonClientHelper = null;
	
	protected String doHandleRequest(Map<String, String> params, Map<String, String> ctx) {
		long startMs = System.currentTimeMillis();
		
		if (params == null) {
			params = new HashMap<String, String>(0);
		}
		
		if (ctx == null) {
			ctx = new HashMap<String, String>(0);
		}
		
		String method = params.get(CommonConfigConstants.THRIFT_PARAM_METHOD);
		
		handleCtx(method, ctx);
		
		ResponseBase response = doRealHandleRequest(params, ctx);
		if (response == null) {
			response = new ResponseBase(CommonErrorCodeConstants.ERR_CODE_THRIFT_RESPONSE_NULL);
		}
		String responseStr = JSON.toJSONString(response);
		
		long timeConsumeMs = System.currentTimeMillis() - startMs;
		
		LOGGER_THRIFT_REQUEST_RESPONSE.info("request ice request[" + chkMapDataLength(params) + "], ctx[" + chkMapDataLength(ctx) + "], response[" + responseStr + "], consume " + timeConsumeMs + " ms");
		
		sendTimeConsumeLog(method, "", ctx, (int)timeConsumeMs, response.getReturnCode());
		return responseStr;
	}
	
	private void handleCtx(String method, Map<String, String> ctx) {
		if (ctx == null) {
			return;
		}
		String flowId = ctx.get(CommonConfigConstants.THRIFT_CTX_FLOW_ID);
		if (StringUtil.isBlank(flowId)) {
			flowId = UUIDUtil.generateUUID();
			ctx.put(CommonConfigConstants.THRIFT_CTX_FLOW_ID, flowId);
		}
		
		String seqNo = ctx.get(CommonConfigConstants.THRIFT_CTX_SEQ_NO);
		if (StringUtil.isBlank(seqNo)) {
			ctx.put(CommonConfigConstants.THRIFT_CTX_SEQ_NO, "" + 1);
		} else {
			try {
				int iSeqNo = Integer.parseInt(seqNo.trim());
				iSeqNo++;
				ctx.put(CommonConfigConstants.THRIFT_CTX_SEQ_NO, "" + iSeqNo);
			} catch (Exception e) {
				LOGGER.error("parse seqNo [" + seqNo + "] to int exception: ", e);
			}
		}
		
		String seq = ctx.get(CommonConfigConstants.THRIFT_CTX_SEQ);
		if (StringUtil.isBlank(seq)) {
			seq = method;
		} else {
			seq = seq + "|" + method;
		}
		ctx.put(CommonConfigConstants.THRIFT_CTX_SEQ, seq);
	}
	
	private ResponseBase doRealHandleRequest(Map<String, String> params, Map<String, String> ctx) {
		if (params == null) {
			params = new HashMap<String, String>(0);
		}
		
		if (ctx == null) {
			ctx = new HashMap<String, String>(0);
		}
		
		String method = params.get(CommonConfigConstants.THRIFT_PARAM_METHOD);
		
		if (CommonConfigConstants.THRIFT_PARAM_METHOD_VAL_KEEP_ALIVE.equalsIgnoreCase(method)) {
			return new ResponseBase(CommonErrorCodeConstants.ERR_CODE_SUCCESS);
		}
		
		if (StringUtil.isBlank(method)) {
			return new ResponseBase(CommonErrorCodeConstants.ERR_CODE_LACK_PARAM, "lack param: " + CommonConfigConstants.THRIFT_PARAM_METHOD);
		}
		
		String flowId = ctx.get(CommonConfigConstants.THRIFT_CTX_FLOW_ID);
		return doHandleApiJsonRequest(method, params, ctx, flowId);
	}
	
	/**
	 * @param method 当前执行的方法
	 * @param invokedMethod 被调用的方法
	 * @param ctx
	 * @param consumeMs
	 * @param resultCode
	 */
	private void sendTimeConsumeLog(String method, String invokedMethod, Map<String, String> ctx, int consumeMs, int resultCode) {
		if (ctx == null) {
			return ;
		}
		
		if (StringUtil.isBlank(thriftConsumeLogKafkaTopic)) {
			return ;
		}
		
		LogThriftTimeConsume logTimeConsume = new LogThriftTimeConsume();
		logTimeConsume.setCost(consumeMs);
		
		String flowId = ctx.get(CommonConfigConstants.THRIFT_CTX_FLOW_ID);
		logTimeConsume.setFlowId(flowId);
		
		logTimeConsume.setTime(System.currentTimeMillis());
		
		logTimeConsume.setRunMethod(method);
		logTimeConsume.setInvokedMethod(invokedMethod);
		
		String seq = ctx.get(CommonConfigConstants.THRIFT_CTX_SEQ);
		logTimeConsume.setSeq(seq);
		
		String seqNo = ctx.get(CommonConfigConstants.THRIFT_CTX_SEQ_NO);
		int iSeqNo = 0;
		if (StringUtil.isBlank(seqNo)) {
			iSeqNo = 1;
		} else {
			try {
				iSeqNo = Integer.parseInt(seqNo.trim());
			} catch (Exception e) {
				LOGGER.error("parse seqNo [" + seqNo + "] to int exception: ", e);
			}
		}
		
		logTimeConsume.setSeqNo(iSeqNo);
		
		InetAddress addr = null;
		try {
			addr = InetAddress.getLocalHost();
		} catch (UnknownHostException e) {
			LOGGER.error("get local host ip exception: ", e);
		}
		if (addr != null) {
			logTimeConsume.setHost(addr.getHostAddress().toString());
		} else {
			logTimeConsume.setHost("");
		}
		
		logTimeConsume.setResult(resultCode);
		
		KafkaStringProducer.sendMsg(thriftConsumeLogKafkaTopic, "" + System.currentTimeMillis(), JSON.toJSONString(logTimeConsume));
		
		LOGGER_THRIFT_CONSUME.info(JSON.toJSONString(logTimeConsume));
	}
	
	/**
	 * 调用Thrift服务
	 * @param fromMethod 调用者的method
	 * @param callback
	 * @param thriftServiceIdentity  
	 * @param parameters 
	 * @param ctx
	 * @param invokeTimeoutMs  
	 * @return
	 */
	public void asyncInvokeThriftService(String fromMethod, AbstractApiJsonCallback callback, String thriftServiceIdentity, Map<String, String> parameters, Map<String, String> ctx, int invokeTimeoutMs) {
		if (callback != null) {
			callback.setFromMethod(fromMethod);
		}
		if (this.thriftApiJsonClientHelper == null)  {
			if (callback != null) {
				String response = JSON.toJSONString(new ResponseBase(CommonErrorCodeConstants.ERR_CODE_SPRING_INNER_SERVICE_FAILED, "thriftApiJsonClientHelper is null"));
				callback.doResponseApiJson(response);
			}
			return;
		}
		
		this.thriftApiJsonClientHelper.asyncInvokeThriftService(callback, thriftServiceIdentity, parameters, ctx, invokeTimeoutMs, fromMethod);
	}
	
	private Map<String, String> chkMapDataLength(Map<String, String> dataMap) {
		Map<String, String> ret = new HashMap<String, String>(0);
		if (dataMap != null) {
			for (Iterator<String> iterator = dataMap.keySet().iterator(); iterator.hasNext();) {
				String key = (String) iterator.next();
				String val = dataMap.get(key);
				if (StringUtil.isNotBlank(val)) {
					String newVal = chkStringDataLength(val);
					ret.put(key, newVal);
				} else {
					ret.put(key, val);
				}
			}
		}
		
		return ret;
	}
	
	private String chkStringDataLength(String s) {
		if (s == null) {
			return "";
		}
		if (s.length() > 200) {
			return s.substring(0, 99) + "...";
		}
		return s;
	}
	
	public String getThriftConsumeLogKafkaTopic() {
		return thriftConsumeLogKafkaTopic;
	}

	public void setThriftConsumeLogKafkaTopic(String thriftConsumeLogKafkaTopic) {
		this.thriftConsumeLogKafkaTopic = thriftConsumeLogKafkaTopic;
	}
	
	public ThriftApiJsonClientHelper getThriftApiJsonClientHelper() {
		return thriftApiJsonClientHelper;
	}

	public void setThriftApiJsonClientHelper(ThriftApiJsonClientHelper thriftApiJsonClientHelper) {
		this.thriftApiJsonClientHelper = thriftApiJsonClientHelper;
	}

	protected abstract ResponseBase doHandleApiJsonRequest(String method, Map<String, String> params, Map<String, String> ctx, String flowId);
	
}
